package com.isyscore.os.etl.model.flinkviewable;

import com.isyscore.os.etl.constant.JobConstant;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

/**
 * @author : zhan9yn
 * @version : 1.0
 * @date : 2021/12/14 10:41 上午
 */
public class FlinkSQLTemplate {

    public static final String CREATE_TABLE = "CREATE TABLE %s (\n" +
            "    %s \n" +
            ")";

    private static Map<Integer, String> SQL_TEMPLATE = Collections.unmodifiableMap(new HashMap(){{
        put(0, "WITH (\n" +
                "    'connector' = 'mysql-x',\n" +
                "    'url' = 'jdbc:mysql://%s:%s/%s?characterEncoding=UTF-8',\n" +
                "    'username' = '%s',\n" +
                "    'password' = '%s',\n" +
                "    'table-name' = '%s'\n" +
                ");\n");
        put(1,"WITH (\n" +
                "    'connector' = 'oracle',\n" +
                "    'url' = 'jdbc:oracle:thin:@%s:%s:%s',\n" +
                "    'username' = '%s',\n" +
                "    'password' = '%s',\n" +
                "    'table-name' = '%s'\n" +
                ");\n");//"oracle"
        put(3, "WITH (\n" +
                "    'connector'='sqlserver-x',\n" +
                "    'url' = 'jdbc:sqlserver://%s:%s;databaseName=%s;useLOBs=false',\n" +
                "    'username'='%s',\n" +
                "    'password'='%s',\n" +
                "    'schema'='%s',\n" +
                "    'druid.validation-query'='select 1',\n" +
                "    'table-name'='%s'\n" +
                ");\n");//"sqlserver"
        put(2,"WITH (\n" +
                "    'connector' = 'clickhouse-x',\n" +
                "    'url' = 'jdbc:clickhouse://%s:%s/%s',\n" +
                "    'username' = '%s',\n" +
                "    'password' = '%s',\n" +
                "    'lookup.cache-type' = 'lru',\n" +
                "    'table-name' = '%s'\n" +
                ");\n");//"clickhouse"
        put(4, "WITH (\n" +
                "    'connector' = 'dm-x',\n" +
                "    'url' = 'jdbc:dm://%s:%s',\n" +
                "    'schema' = '%s',\n" +
                "    'username' = '%s',\n" +
                "    'password' = '%s',\n" +
                "    'scan.fetch-size' = '2',\n" +
                "    'scan.query-timeout' = '10',\n" +
                "    'lookup.cache-type' = 'lru',\n" +
                "    'table-name' = '%s'\n" +
                ");\n");//"dm"
        put(5, "");//"kafka"

        put(JobConstant.DATA_SYNC_TYPE, "insert into %s\n" +
                "select * from %s;\n");
    }});

    public static String getSQL(Integer type) {
        return SQL_TEMPLATE.get(type);
    }

}
